/*
 * Copyright (C) 2015   Jeremy Chen jeremy_cz@yahoo.com
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *      http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */

#include <utils/CFdbUDPSession.h>
#include <common_base/CFdbSessionContainer.h>
#include <common_base/CFdbContext.h>
#include <common_base/CLogProducer.h>
#include <common_base/CSocketImp.h>
#include <common_base/CFdbMessage.h>
#include <platform/socket/linux/CLinuxSocket.h>
#include <utils/Log.h>
#include <utils/CFdbIfMessageHeader.h>

CFdbUDPSession::CFdbUDPSession(CFdbSessionContainer *container, CSocketImp *socket)
    : CBaseSession(container, socket)
{
}

CFdbUDPSession::~CFdbUDPSession()
{
    mContainer->mUDPSession = 0;
    if (mSocket)
    {
        delete mSocket;
        mSocket = 0;
    }
    descriptor(0);
}

bool CFdbUDPSession::sendMessage(CFdbMessage *msg)
{
    if (!msg->buildHeader(this))
    {
        return false;
    }
    doStatistics(msg->type(), msg->flag(), mStatistics.mTx);
    if (mSocket->send(msg->getRawBuffer(), msg->getRawDataSize()))
    {
        if (msg->isLogEnabled())
        {
            auto logger = FDB_CONTEXT->getLogger();
            if (logger)
            {
                logger->logFDBus(msg, 0, mContainer->owner(), qos());
            }
        }
        return true;
    }
    return false;
}

bool CFdbUDPSession::receiveData(uint8_t *buf, int32_t size)
{
    return mSocket->recv(buf, size) >= 0;
}

void CFdbUDPSession::onError()
{
    onHup();
}

void CFdbUDPSession::onHup()
{
    delete this;
}

void CFdbUDPSession::doBroadcast(NFdbBase::CFdbMessageHeader &head)
{
    auto msg = new CFdbMessage(head, this);
    auto object = mContainer->owner()->getObject(msg, false);
    CBaseJob::Ptr msg_ref(msg);
    if (object)
    {
        msg->decodeDebugInfo(head);
        object->doBroadcast(msg_ref);
    }
}

void CFdbUDPSession::doRequest(NFdbBase::CFdbMessageHeader &head)
{
    auto msg = new CFdbMessage(head, this);
    auto object = mContainer->owner()->getObject(msg, true);
    CBaseJob::Ptr msg_ref(msg);

    if (object)
    {
        msg->decodeDebugInfo(head);
        switch (head.type())
        {
            case FDB_MT_REQUEST:
                if (mContainer->owner()->onMessageAuthentication(msg))
                {
                    object->doInvoke(msg_ref);
                }
                else
                {
                }
            break;
            case FDB_MT_PUBLISH:
                if (mContainer->owner()->onEventAuthentication(msg))
                {
                    object->doPublish(msg_ref);
                }
                else
                {
                }
            break;
            default:
            break;
        }
    }
    else
    {
    }
}

void CFdbUDPSession::setUDPDestAddr(const char *ip_addr, int32_t port)
{
    auto udp_socket_imp = fdb_dynamic_cast_if_available<CUDPTransportSocket *>(mSocket);
    udp_socket_imp->setUDPDestAddr(ip_addr, port);
}

const std::string &CFdbUDPSession::getUDPDestAddr(int32_t &port) const
{
    auto udp_socket_imp = fdb_dynamic_cast_if_available<CUDPTransportSocket *>(mSocket);
    return udp_socket_imp->getUDPDestAddr(port);
}

EFdbQOS CFdbUDPSession::qos()
{
    return mContainer->isSecure() ? FDB_QOS_SECURE_BEST_EFFORTS : FDB_QOS_BEST_EFFORTS;
}

